package com.shujia.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator

object Demo19Acc {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("pi")
      .setMaster("local")

    val sc = new SparkContext(conf)


    val studentRDD: RDD[String] = sc.textFile("spark/data/students.txt")


    /**
      * 累加器
      *
      * 累加过程
      * 1  在每一个Executor进行局部累加
      * 2  当job执行完成之后数据拉取到Driver端进行汇总
      *
      *
      */

    //    var i = 0
    //
    //    studentRDD.foreach(line => {
    //      i += 1
    //    })
    //
    //    println(i)


    //1 在Driver端定义一个累加器的变量
    val acc: LongAccumulator = sc.longAccumulator


    studentRDD.foreach(line => {

      //在Executor端进行累加
      //不能在Executor端读取累加器的结果
      acc.add(1)
    })


    //读取累加结果
    println(acc.value)

  }

}
